﻿using BaiTu.IQueueService.Providers;
using BaiTu.QueueEntity;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace BaiTu.QueueService.Providers
{
    class RedisProvider : IQueueProvider
    {
        /// <summary>
        /// 永远返回true
        /// </summary>
        /// <param name="queueName"></param>
        /// <returns></returns>
        public bool IsQueueExists(string queueName)
        {
            return BaiTu.RedisExecutors.RedisExecutor.ContainsKey(queueName);
        }

        /// <summary>
        /// 建立队列
        /// </summary>
        /// <param name="queueName"></param>
        public void CreateQueue(string queueName)
        {
            //不作任何事情
            //Redis会自动建立队列
        }

        public void DeleteQueue(string queueName)
        {
            BaiTu.RedisExecutors.RedisExecutor.Remove(queueName);
        }

        public string SendMessage(string queueName, InnerMessageWrapper msg)
        {
            string msgID = "";
            string listId = string.Format(QueueConfiguration.RedisConfigInfo.QueueNameFormat, queueName);

            if (string.IsNullOrWhiteSpace(msg.MessageID))
            {
                msg.MessageID = UniqueIdGenerator.IDGenerator.Next().ToString();
            }
            else
            {
                msg.MessageID = msg.MessageID;
            }
            msgID = msg.MessageID;

            byte[] bytes = Common.Helpers.XMLSerializationHelper.Serialize2Bytes_Protobuf<InnerMessageWrapper>(msg);
            BaiTu.RedisExecutors.RedisExecutor.LPUSH(listId, bytes);
            return msgID;
        }

        public InnerMessageWrapper ReceiveMessage(string queueName, string relatedMsgId = "")
        {
            InnerMessageWrapper obj = null;

            string listId = string.Format(QueueConfiguration.RedisConfigInfo.QueueNameFormat, queueName);
            string lockId = string.Format("{0}.lock", listId);


            if (string.IsNullOrEmpty(relatedMsgId))
            {
                byte[] bytes =BaiTu.RedisExecutors.RedisExecutor.BRPopValue(listId, 0);
                if (bytes != null)
                    obj = Common.Helpers.XMLSerializationHelper.DeserializeFromBytes_Protobuf<InnerMessageWrapper>(bytes);
            }
            else
            {
                while (true)
                {
                    byte[] bytes = BaiTu.RedisExecutors.RedisExecutor.BRPopValue(listId, 0);

                    if (bytes == null)
                        continue;

                    if (bytes != null)
                        obj = Common.Helpers.XMLSerializationHelper.DeserializeFromBytes_Protobuf<InnerMessageWrapper>(bytes);

                    if (obj == null)
                        continue;

                    if (obj.CorrelationID != relatedMsgId)
                        continue;

                    BaiTu.RedisExecutors.RedisExecutor.LPUSH(queueName, bytes);
                    break;
                }
            }


            return obj;
        }
    }
}
